package main

// https://blog.golang.org/pipelines

import (
	"net/http"
	"crypto/md5"
	"fmt"
	"os"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/ngaut/log"

	// import写的是package路径
	utils "gitee.com/cshi/go-learning/concurrency/src"
)

func Goid() int {
	defer func() {
		if err := recover(); err != nil {
			fmt.Printf("panic recover:panic info:%v", err)
		}
	}()

	var buf [64]byte
	n := runtime.Stack(buf[:], false)
	idField := strings.Fields(strings.TrimPrefix(string(buf[:n]), "goroutine "))[0]
	id, err := strconv.Atoi(idField)
	if err != nil {
		panic(fmt.Sprintf("cannot get goroutine id: %v", err))
	}
	return id
}

func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

func sq(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		//id := Goid()
		// 对channel进行range loop，在最后一个元素后就会退出，比起用 c, ok <- in 的方式要好 ！
		for n := range in {
			out <- n * n
			//fmt.Printf("%v n=%v\n", id, n)
		}
		close(out)
	}()
	return out
}

func tSimple1() {
	c := gen(2, 3)
	out := sq(c)

	// consume
	fmt.Println(<-out)
	fmt.Println(<-out)

	// 由于inbound, outbound channel有相同类型，所以可以compose
	for n := range sq(sq(gen(2, 3))) {
		fmt.Println(n)
	}
}

//////////////////
func merge(cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	output := func(c <-chan int) {
		for n := range c {
			out <- n
		}
		wg.Done()
	}
	wg.Add(len(cs))

	for _, c := range cs {
		go output(c)
	}

	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

// 相同的输入由多个输出来接收，此为fan-out
// 这里面存在的问题是：若receiver不想接受所有的sent value呢
func t2SimpleWithMerge() {
	in := gen(2, 3)
	// 发现将in分给两个consumer，似乎不会发生某个consumer得不到数据的情况
	c1 := sq(in)
	c2 := sq(in)
	for n := range merge(c1, c2) {
		log.Infof("%v", n)
	}
}

//////////////////
func mergeWithDoneChannel(done <-chan struct{}, cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	output := func(c <-chan int, id int) {
		for n := range c {
			log.Infof("%v, n=%v", id, n)
			select {
			// 这个意思难道在说，如果done中有值，一定走done，否则才写out ？？
			case out <- n:
				log.Infof("%v: sender send %v", id, n)
			case <-done:
				log.Infof("%v: sender goroutine recv done", id)
			}
			log.Infof("%v: sender read value: %v", id, n)
		}
		log.Infof("%v: sender ready to exit", id)
		wg.Done()
	}

	wg.Add(len(cs))

	for i, c := range cs {
		go output(c, i)
	}

	go func() {
		wg.Wait()
		log.Infof("found all sender exited. close out channel")
		close(out)
	}()

	return out
}
func t3ReceiverReturnPrematurely() {
	in := gen(4, 5, 6)
	c1 := sq(in)
	c2 := sq(in)

	done := make(chan struct{}, 2) // 实际上不容易确定这个大小 ， 这是问题！
	out := mergeWithDoneChannel(done, c1, c2)
	fmt.Println(<-out)

	// 当receiver收到一个元素后就想返回时，可以给done发信号
	log.Infof("before sending the first done signal")
	done <- struct{}{}
	time.Sleep(2 * time.Second)
	log.Infof("before sending the second done signal")
	done <- struct{}{}
	time.Sleep(2 * time.Second)

	fmt.Println("t3 exits")
}

////////////// 把done channel传给所有的goroutine，以确保当接收方不再接收数据时，sender可以停止
func genWithDone(done <-chan struct{}, nums ...int) <-chan int {
	out := make(chan int)

	go func() {
		// 要保证关闭，否则接收方一直接收到0
		defer close(out)
		for _, n := range nums {
			select {
			case out <- n:
			case <-done:
				return
			}
		}
	}()

	return out
}
func sqWithDone(done <-chan struct{}, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case out <- n * n:
			case <-done:
				return
			}
		}
	}()
	return out
}
func mergeReturnOnReceDone(done <-chan struct{}, cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	output := func(c <-chan int) {
		defer wg.Done()
		// out在这里要不要关闭呢？？？
		for n := range c {
			select {
			case out <- n:
			case <-done:
				return
			}
		}
	}

	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

func t4PassDoneChannelToAllGoroutine() {
	done := make(chan struct{})
	defer close(done)

	in := genWithDone(done, 2, 3, 4)
	c1 := sqWithDone(done, in)
	c2 := sqWithDone(done, in)

	out := mergeReturnOnReceDone(done, c1, c2)
	fmt.Println(<-out)

	// done一关闭，all sender will be notified, so they will exit
}

func t5md5Calculate(root string, parallel bool) {
	var err error
	var m map[string][md5.Size]byte

	done := make(chan bool)
	defer close(done)

	t := time.NewTimer(1*time.Second)
	go func ()  {
		for {
			select {
			case <-t.C:
				fmt.Println(runtime.NumGoroutine())
			case <-done:
				//// 在这打印之前 ，main goroutine可能已经退出，所以它可能没机会打印
				log.Infof("goroutine of calculating number of goroutines exits")
				return
			}

			t.Reset(1*time.Second)
		}
	}()

	start := time.Now()
	if parallel {
		m, err = utils.Tmd5AllParallel(root)
	} else {
		m, err = utils.Tmd5AllSerial(root)
	}

	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Printf("parallel: %v, md5 took: %v\n", parallel, time.Since(start))

	var paths []string
	for path := range m {
		paths = append(paths, path)
	}

	sort.Strings(paths)
	// for _, path := range paths {
	// 	// 模拟md5sum 的输出
	// 	fmt.Printf("%x  %s\n", m[path], path)
	// }
}

func main() {
	// 像这样的协程，是长期后台协程，应该不用考虑协程Leak 吧
	go func ()  {
		fmt.Println(http.ListenAndServe(":9876", nil))
	}()

	//tSimple1()
	//t2SimpleWithMerge()
	//t3ReceiverReturnPrematurely()
	//t4PassDoneChannelToAllGoroutine()

	t5md5Calculate(os.Args[1], true)
	t5md5Calculate(os.Args[1], false)
}
